package org.liuwei.rxjava;

import org.junit.Test;

import rx.Observer;
import rx.functions.Action1;
import rx.subjects.AsyncSubject;
import rx.subjects.BehaviorSubject;
import rx.subjects.PublishSubject;
import rx.subjects.ReplaySubject;

/**
 * @see https://my.oschina.net/u/1266221/blog/952634
 * @Description: RXjava 基本使用示例
 * @author wei.liu
 * @date 2017年6月14日 下午1:58:20
 * @version V1.0
 */
public class demo3 {

	// Observer只会接收asyncSubject的onCompleted()被调用前的最后一个数据，即“asyncSubject3”，
	// 如果不调用onCompleted()，Subscriber将不接收任何数据。
	@Test
	public void testAsyncSubject() {
		AsyncSubject<String> asyncSubject = AsyncSubject.create();
		asyncSubject.onNext("asyncSubject1");
		asyncSubject.onNext("asyncSubject2");
		asyncSubject.onNext("asyncSubject3");
		asyncSubject.onCompleted();
		asyncSubject.subscribe(new Observer<String>() {
			@Override
			public void onCompleted() {

				System.out.println("asyncSubject onCompleted"); // 输出
																// asyncSubject
																// onCompleted
			}

			@Override
			public void onError(Throwable e) {

				System.out.println("asyncSubject onError"); // 不输出（异常才会输出）
			}

			@Override
			public void onNext(String s) {

				System.out.println("asyncSubject:" + s); // 输出asyncSubject:asyncSubject3
			}
		});
	}

	@Test
	public void testBehaviorSubject() {
		BehaviorSubject<String> behaviorSubject = BehaviorSubject.create("default");
		behaviorSubject.onNext("behaviorSubject1");
		behaviorSubject.onNext("behaviorSubject2");
		behaviorSubject.subscribe(new Observer<String>() {
			@Override
			public void onCompleted() {

				System.out.println("behaviorSubject:complete");
			}

			@Override
			public void onError(Throwable e) {

				System.out.println("behaviorSubject:error");
			}

			@Override
			public void onNext(String s) {

				System.out.println("behaviorSubject:" + s);
			}
		});

		behaviorSubject.onNext("behaviorSubject3");
		behaviorSubject.onNext("behaviorSubject4");
	}

	@Test
	public void testPublishSubject() {
		PublishSubject<String> publishSubject = PublishSubject.create();
		publishSubject.onNext("publishSubject1");
		publishSubject.onNext("publishSubject2");
		publishSubject.subscribe(new Observer<String>() {
			@Override
			public void onCompleted() {

			}

			@Override
			public void onError(Throwable e) {

			}

			@Override
			public void onNext(String s) {
				System.out.println("publishSubject observer1:" + s);
			}
		});
		publishSubject.onNext("publishSubject3");
		publishSubject.onNext("publishSubject4");
	}

	@Test
	public void testReplaySubject() {
		ReplaySubject<String> replaySubject = ReplaySubject.create(); // 创建默认初始缓存容量大小为16的ReplaySubject，当数据条目超过16会重新分配内存空间，使用这种方式，不论ReplaySubject何时被订阅，Observer都能接收到数据
		// replaySubject =
		// ReplaySubject.create(100);//创建指定初始缓存容量大小为100的ReplaySubject
		// replaySubject = ReplaySubject.createWithSize(2);//只缓存订阅前最后发送的2条数据
		// replaySubject=ReplaySubject.createWithTime(1,TimeUnit.SECONDS,Schedulers.computation());
		// //replaySubject被订阅前的前1秒内发送的数据才能被接收
		replaySubject.onNext("replaySubject:pre1");
		replaySubject.onNext("replaySubject:pre2");
		replaySubject.onNext("replaySubject:pre3");
		replaySubject.subscribe(new Action1<String>() {
			@Override
			public void call(String s) {
				System.out.println("replaySubject:" + s);
			}
		});
		replaySubject.onNext("replaySubject:after1");
		replaySubject.onNext("replaySubject:after2");
	}

}
